//! 集群网络管理模块
//! 
//! 实现Redis集群的双端口网络架构:
//! 1. Redis客户端端口 (如6379) - 处理客户端请求和数据读写
//! 2. 集群内部通信端口 (如16379) - 处理Gossip协议、故障检测、集群状态同步

use crate::cluster::traits::{
    NetworkManager, NetworkEvent, NetworkMessage, NetworkEventHandler, NetworkStats
};
use crate::cluster::{ClusterConfig, ClusterError, ClusterResult};
use async_trait::async_trait;
use std::collections::HashMap;
use std::net::SocketAddr;
use std::sync::Arc;
use std::time::Instant;
use tokio::net::{TcpListener, TcpStream};
use tokio::sync::mpsc::{self, UnboundedReceiver, UnboundedSender};
use tokio::sync::{broadcast, RwLock};
use tracing::{debug, error, info, warn};

/// 网络消息类型
#[derive(Debug, Clone)]
pub enum ClusterNetworkMessage {
    /// 客户端连接
    ClientConnected {
        addr: SocketAddr,
    },
    /// 客户端断开
    ClientDisconnected { addr: SocketAddr },
    /// 集群节点连接
    ClusterNodeConnected {
        addr: SocketAddr,
    },
    /// 集群节点断开
    ClusterNodeDisconnected { addr: SocketAddr },
    /// 网络错误
    NetworkError { error: String, addr: Option<SocketAddr> },
    /// 原始数据消息
    RawMessage {
        id: String,
        message_type: String,
        payload: Vec<u8>,
    },
}

impl NetworkMessage for ClusterNetworkMessage {
    fn message_id(&self) -> &str {
        match self {
            Self::ClientConnected { .. } => "client_connected",
            Self::ClientDisconnected { .. } => "client_disconnected",
            Self::ClusterNodeConnected { .. } => "cluster_connected",
            Self::ClusterNodeDisconnected { .. } => "cluster_disconnected",
            Self::NetworkError { .. } => "network_error",
            Self::RawMessage { id, .. } => id,
        }
    }
    
    fn message_type(&self) -> &str {
        match self {
            Self::ClientConnected { .. } => "client_connected",
            Self::ClientDisconnected { .. } => "client_disconnected",
            Self::ClusterNodeConnected { .. } => "cluster_connected",
            Self::ClusterNodeDisconnected { .. } => "cluster_disconnected",
            Self::NetworkError { .. } => "network_error",
            Self::RawMessage { message_type, .. } => message_type,
        }
    }
    
    fn payload(&self) -> &[u8] {
        match self {
            Self::RawMessage { payload, .. } => payload,
            _ => &[],
        }
    }
    
    fn set_payload(&mut self, new_payload: Vec<u8>) {
        match self {
            Self::RawMessage { payload, .. } => *payload = new_payload,
            _ => {} // 其他类型不支持设置payload
        }
    }
    
    fn serialize(&self) -> ClusterResult<Vec<u8>> {
        // 手动序列化，避免TcpStream序列化问题
        match self {
            Self::ClientConnected { addr } => {
                let data = format!("ClientConnected:{addr}");
                Ok(data.into_bytes())
            }
            Self::ClientDisconnected { addr } => {
                let data = format!("ClientDisconnected:{addr}");
                Ok(data.into_bytes())
            }
            Self::ClusterNodeConnected { addr } => {
                let data = format!("ClusterNodeConnected:{addr}");
                Ok(data.into_bytes())
            }
            Self::ClusterNodeDisconnected { addr } => {
                let data = format!("ClusterNodeDisconnected:{addr}");
                Ok(data.into_bytes())
            }
            Self::NetworkError { error, addr } => {
                let data = format!("NetworkError:{error}:{addr:?}");
                Ok(data.into_bytes())
            }
            Self::RawMessage { id, message_type, payload } => {
                let payload_str = String::from_utf8_lossy(payload);
                let data = format!("RawMessage:{id}:{message_type}:{payload_str}");
                Ok(data.into_bytes())
            }
        }
    }
    
    fn deserialize(data: &[u8]) -> ClusterResult<Self> {
        let data_str = String::from_utf8(data.to_vec())
            .map_err(|e| ClusterError::Config(format!("解析消息失败: {e}")))?;
        
        let parts: Vec<&str> = data_str.splitn(2, ':').collect();
        if parts.len() != 2 {
            return Err(ClusterError::Config("消息格式无效".to_string()));
        }
        
        match parts[0] {
            "ClientConnected" => {
                let addr = parts[1].parse()
                    .map_err(|e| ClusterError::Config(format!("地址解析失败: {e}")))?;
                Ok(Self::ClientConnected { addr })
            }
            "ClientDisconnected" => {
                let addr = parts[1].parse()
                    .map_err(|e| ClusterError::Config(format!("地址解析失败: {e}")))?;
                Ok(Self::ClientDisconnected { addr })
            }
            "ClusterNodeConnected" => {
                let addr = parts[1].parse()
                    .map_err(|e| ClusterError::Config(format!("地址解析失败: {e}")))?;
                Ok(Self::ClusterNodeConnected { addr })
            }
            "ClusterNodeDisconnected" => {
                let addr = parts[1].parse()
                    .map_err(|e| ClusterError::Config(format!("地址解析失败: {e}")))?;
                Ok(Self::ClusterNodeDisconnected { addr })
            }
            _ => Err(ClusterError::Config("未知消息类型".to_string()))
        }
    }
}

/// 网络事件类型
#[derive(Debug, Clone)]
pub struct ClusterNetworkEvent {
    pub source_node: String,
    pub event_type: String,
    pub timestamp: Instant,
    pub data: Option<Vec<u8>>,
}

impl NetworkEvent for ClusterNetworkEvent {
    fn source_node(&self) -> &str {
        &self.source_node
    }
    
    fn event_type(&self) -> &str {
        &self.event_type
    }
    
    fn timestamp(&self) -> Instant {
        self.timestamp
    }
}

/// 网络事件处理器trait
#[async_trait]
pub trait ClusterNetworkEventHandler: Send + Sync {
    /// 处理客户端连接
    async fn handle_client_connection(&self, stream: Arc<RwLock<TcpStream>>, addr: SocketAddr);
    
    /// 处理集群节点连接
    async fn handle_cluster_connection(&self, stream: Arc<RwLock<TcpStream>>, addr: SocketAddr);
    
    /// 处理连接断开
    async fn handle_disconnection(&self, addr: SocketAddr, is_cluster: bool);
    
    /// 处理网络错误
    async fn handle_network_error(&self, error: String, addr: Option<SocketAddr>);
}

/// 集群网络管理器
pub struct ClusterNetworkManager {
    /// 配置信息
    config: ClusterConfig,
    
    /// Redis客户端监听器
    redis_listener: Option<TcpListener>,
    
    /// 集群通信监听器
    cluster_listener: Option<TcpListener>,
    
    /// 网络消息发送器
    message_sender: UnboundedSender<ClusterNetworkMessage>,
    
    /// 网络消息接收器
    message_receiver: Option<UnboundedReceiver<ClusterNetworkMessage>>,
    
    /// 广播发送器 (用于向所有处理器广播事件)
    broadcast_sender: broadcast::Sender<ClusterNetworkMessage>,
    
    /// 事件处理器
    event_handler: Option<Box<dyn NetworkEventHandler<Event = ClusterNetworkEvent>>>,
    
    /// 运行状态
    is_running: Arc<RwLock<bool>>,
    
    /// 网络统计信息
    stats: Arc<RwLock<NetworkStats>>,
    
    /// 连接状态管理
    connections: Arc<RwLock<HashMap<String, bool>>>,
}

impl ClusterNetworkManager {
    /// 创建新的网络管理器
    pub async fn new(config: ClusterConfig) -> ClusterResult<Self> {
        // 验证配置
        config.validate().map_err(ClusterError::Config)?;
        
        // 创建消息通道
        let (message_sender, message_receiver) = mpsc::unbounded_channel();
        let (broadcast_sender, _) = broadcast::channel(1000);
        
        let stats = NetworkStats {
            total_connections: 0,
            active_connections: 0,
            messages_sent: 0,
            messages_received: 0,
            bytes_sent: 0,
            bytes_received: 0,
            connection_errors: 0,
            message_errors: 0,
        };
        
        Ok(Self {
            config,
            redis_listener: None,
            cluster_listener: None,
            message_sender,
            message_receiver: Some(message_receiver),
            broadcast_sender,
            event_handler: None,
            is_running: Arc::new(RwLock::new(false)),
            stats: Arc::new(RwLock::new(stats)),
            connections: Arc::new(RwLock::new(HashMap::new())),
        })
    }
    

    
    /// 绑定监听端口
    pub async fn bind(&mut self) -> ClusterResult<()> {
        // 只绑定集群通信端口，Redis客户端端口由独立的Redis服务器处理
        let cluster_addr = self.config.cluster_socket_addr()
            .map_err(|e| ClusterError::Config(format!("无效的集群地址: {e}")))?;
        
        let cluster_listener = TcpListener::bind(&cluster_addr).await
            .map_err(ClusterError::Network)?;
        
        info!("集群通信端口监听在: {}", cluster_addr);
        self.cluster_listener = Some(cluster_listener);
        
        Ok(())
    }
    
    /// 启动网络服务
    pub async fn start(&mut self) -> ClusterResult<()> {
        if self.cluster_listener.is_none() {
            return Err(ClusterError::Config("请先调用bind()绑定端口".to_string()));
        }
        
        {
            let mut running = self.is_running.write().await;
            if *running {
                return Err(ClusterError::Config("网络服务已在运行".to_string()));
            }
            *running = true;
        }
        
        info!("启动集群网络服务...");
        
        // 启动消息处理任务
        self.start_message_processor().await?;
        
        // 只启动集群通信监听任务，Redis客户端由单独的服务器处理
        self.start_cluster_listener().await?;
        
        info!("集群网络服务启动成功");
        Ok(())
    }
    
    /// 停止网络服务
    pub async fn stop(&mut self) -> ClusterResult<()> {
        let mut running = self.is_running.write().await;
        if !*running {
            return Ok(());
        }
        
        *running = false;
        info!("正在停止集群网络服务...");
        
        // 关闭监听器
        if let Some(listener) = self.redis_listener.take() {
            drop(listener);
        }
        
        if let Some(listener) = self.cluster_listener.take() {
            drop(listener);
        }
        
        info!("集群网络服务已停止");
        Ok(())
    }
    
    /// 检查服务是否运行中
    pub async fn is_running(&self) -> bool {
        *self.is_running.read().await
    }
    
    /// 获取配置信息
    pub fn config(&self) -> &ClusterConfig {
        &self.config
    }
    
    /// 获取消息发送器
    pub fn message_sender(&self) -> UnboundedSender<ClusterNetworkMessage> {
        self.message_sender.clone()
    }
    
    /// 获取广播接收器
    pub fn subscribe(&self) -> broadcast::Receiver<ClusterNetworkMessage> {
        self.broadcast_sender.subscribe()
    }
}



#[async_trait]
impl NetworkManager for ClusterNetworkManager {
    type Event = ClusterNetworkEvent;
    type Message = ClusterNetworkMessage;
    
    async fn start(&mut self) -> ClusterResult<()> {
        if self.cluster_listener.is_none() {
            return Err(ClusterError::Config("请先调用bind()绑定端口".to_string()));
        }
        
        {
            let mut running = self.is_running.write().await;
            if *running {
                return Err(ClusterError::Config("网络服务已在运行".to_string()));
            }
            *running = true;
        }
        
        info!("启动集群网络服务...");
        
        // 启动消息处理任务
        self.start_message_processor().await?;
        
        // 只启动集群通信监听任务
        self.start_cluster_listener().await?;
        
        info!("集群网络服务启动成功");
        Ok(())
    }
    
    async fn stop(&mut self) -> ClusterResult<()> {
        let mut running = self.is_running.write().await;
        if !*running {
            return Ok(());
        }
        
        *running = false;
        info!("正在停止集群网络服务...");
        
        // 关闭监听器
        if let Some(listener) = self.redis_listener.take() {
            drop(listener);
        }
        
        if let Some(listener) = self.cluster_listener.take() {
            drop(listener);
        }
        
        info!("集群网络服务已停止");
        Ok(())
    }
    
    async fn bind(&mut self) -> ClusterResult<()> {
        // 绑定Redis客户端端口
        let redis_addr = self.config.redis_socket_addr()
            .map_err(|e| ClusterError::Config(format!("无效的Redis地址: {e}")))?;
        
        let redis_listener = TcpListener::bind(&redis_addr).await
            .map_err(ClusterError::Network)?;
        
        info!("Redis客户端端口监听在: {}", redis_addr);
        self.redis_listener = Some(redis_listener);
        
        // 绑定集群通信端口
        let cluster_addr = self.config.cluster_socket_addr()
            .map_err(|e| ClusterError::Config(format!("无效的集群地址: {e}")))?;
        
        let cluster_listener = TcpListener::bind(&cluster_addr).await
            .map_err(ClusterError::Network)?;
        
        info!("集群通信端口监听在: {}", cluster_addr);
        self.cluster_listener = Some(cluster_listener);
        
        Ok(())
    }
    
    async fn send_message(&self, _target: &str, message: Self::Message) -> ClusterResult<()> {
        // TODO: 实现向指定目标发送消息
        self.message_sender.send(message)
            .map_err(|e| ClusterError::Config(format!("发送消息失败: {e}")))?;
        
        // 更新统计
        let mut stats = self.stats.write().await;
        stats.messages_sent += 1;
        
        Ok(())
    }
    
    async fn broadcast_message(&self, message: Self::Message) -> ClusterResult<()> {
        self.broadcast_sender.send(message)
            .map_err(|e| ClusterError::Config(format!("广播失败: {e}")))?;
        
        // 更新统计
        let mut stats = self.stats.write().await;
        stats.messages_sent += 1;
        
        Ok(())
    }
    
    async fn next_event(&mut self) -> Option<Self::Event> {
        // TODO: 实现事件接收
        None
    }
    
    fn set_event_handler(&mut self, handler: Box<dyn NetworkEventHandler<Event = Self::Event>>) {
        self.event_handler = Some(handler);
    }
    
    async fn get_stats(&self) -> NetworkStats {
        self.stats.read().await.clone()
    }
    
    async fn is_connected(&self, node_id: &str) -> bool {
        self.connections.read().await.get(node_id).copied().unwrap_or(false)
    }
    
    async fn connect_to_node(&mut self, node_id: &str, addr: SocketAddr) -> ClusterResult<()> {
        let _stream = TcpStream::connect(addr).await
            .map_err(ClusterError::Network)?;
        
        // 记录连接状态
        self.connections.write().await.insert(node_id.to_string(), true);
        
        // 更新统计
        let mut stats = self.stats.write().await;
        stats.total_connections += 1;
        stats.active_connections += 1;
        
        info!("连接到集群节点: {} ({})", node_id, addr);
        Ok(())
    }
    
    async fn disconnect_from_node(&mut self, node_id: &str) -> ClusterResult<()> {
        if self.connections.write().await.remove(node_id).is_some() {
            // 更新统计
            let mut stats = self.stats.write().await;
            stats.active_connections = stats.active_connections.saturating_sub(1);
            
            info!("断开与集群节点的连接: {}", node_id);
        }
        Ok(())
    }
}

impl ClusterNetworkManager {
    /// 启动消息处理器
    async fn start_message_processor(&mut self) -> ClusterResult<()> {
        let mut receiver = self.message_receiver.take()
            .ok_or_else(|| ClusterError::Config("消息接收器已被使用".to_string()))?;
        
        let broadcast_sender = self.broadcast_sender.clone();
        let is_running = self.is_running.clone();
        
        tokio::spawn(async move {
            while *is_running.read().await {
                match receiver.recv().await {
                    Some(message) => {
                        // 广播消息给所有订阅者
                        if let Err(e) = broadcast_sender.send(message.clone()) {
                            warn!("广播消息失败: {}", e);
                        }
                        
                        // TODO: 处理事件，由于不能克隆trait对象，这里暂时省略
                    }
                    None => {
                        debug!("消息通道已关闭");
                        break;
                    }
                }
            }
        });
        
        Ok(())
    }
    

    
    /// 启动集群通信监听器
    async fn start_cluster_listener(&self) -> ClusterResult<()> {
        let cluster_addr = self.config.cluster_socket_addr()
            .map_err(|e| ClusterError::Config(format!("无效的集群地址: {e}")))?;
        
        let message_sender = self.message_sender.clone();
        let is_running = self.is_running.clone();
        
        tokio::spawn(async move {
            let listener = match TcpListener::bind(cluster_addr).await {
                Ok(listener) => listener,
                Err(e) => {
                    error!("绑定集群监听器失败: {}", e);
                    return;
                }
            };
            
            while *is_running.read().await {
                match listener.accept().await {
                    Ok((_stream, addr)) => {
                        info!("接受集群节点连接: {}", addr);
                        
                        let message = ClusterNetworkMessage::ClusterNodeConnected {
                            addr,
                        };
                        
                        if let Err(e) = message_sender.send(message) {
                            error!("发送集群连接消息失败: {}", e);
                        }
                    }
                    Err(e) => {
                        error!("接受集群节点连接失败: {}", e);
                        
                        let message = ClusterNetworkMessage::NetworkError {
                            error: format!("集群监听器错误: {e}"),
                            addr: None,
                        };
                        
                        if let Err(send_err) = message_sender.send(message) {
                            error!("发送网络错误消息失败: {}", send_err);
                        }
                    }
                }
            }
        });
        
        Ok(())
    }
    
    /// 连接到远程集群节点（原有方法保持兼容）
    pub async fn connect_to_node_old(&self, addr: SocketAddr) -> ClusterResult<Arc<RwLock<TcpStream>>> {
        let stream = TcpStream::connect(addr).await
            .map_err(ClusterError::Network)?;
        
        info!("连接到集群节点: {}", addr);
        Ok(Arc::new(RwLock::new(stream)))
    }
    
    /// 广播消息给所有监听器（原有方法保持兼容）
    pub fn broadcast(&self, message: ClusterNetworkMessage) -> ClusterResult<()> {
        self.broadcast_sender.send(message)
            .map_err(|e| ClusterError::Config(format!("广播失败: {e}")))?;
        Ok(())
    }
    
    /// 发送网络消息（原有方法保持兼容）
    pub fn send_message_old(&self, message: ClusterNetworkMessage) -> ClusterResult<()> {
        self.message_sender.send(message)
            .map_err(|e| ClusterError::Config(format!("发送消息失败: {e}")))?;
        Ok(())
    }
}

impl Drop for ClusterNetworkManager {
    fn drop(&mut self) {
        // 网络管理器销毁时会自动停止所有任务
        // 因为我们使用 Arc<RwLock<bool>> 来控制任务的运行状态
        // 当网络管理器被销毁时，running 标志会自动变为 false
        debug!("集群网络管理器已销毁");
    }
}

/// 默认网络事件处理器 (用于测试和演示)
pub struct DefaultNetworkEventHandler;

#[async_trait]
impl NetworkEventHandler for DefaultNetworkEventHandler {
    type Event = ClusterNetworkEvent;
    
    async fn handle_event(&mut self, event: Self::Event) -> ClusterResult<()> {
        match event.event_type() {
            "connection" => info!("处理连接事件: {}", event.source_node()),
            "disconnection" => info!("处理断开连接事件: {}", event.source_node()),
            "error" => error!("处理错误事件: {}", event.source_node()),
            _ => debug!("处理未知事件类型: {}", event.event_type()),
        }
        Ok(())
    }
}

#[async_trait]
impl ClusterNetworkEventHandler for DefaultNetworkEventHandler {
    async fn handle_client_connection(&self, _stream: Arc<RwLock<TcpStream>>, addr: SocketAddr) {
        info!("处理客户端连接: {}", addr);
        // TODO: 实现客户端连接处理逻辑
    }
    
    async fn handle_cluster_connection(&self, _stream: Arc<RwLock<TcpStream>>, addr: SocketAddr) {
        info!("处理集群节点连接: {}", addr);
        // TODO: 实现集群节点连接处理逻辑
    }
    
    async fn handle_disconnection(&self, addr: SocketAddr, is_cluster: bool) {
        if is_cluster {
            info!("集群节点断开连接: {}", addr);
        } else {
            info!("客户端断开连接: {}", addr);
        }
    }
    
    async fn handle_network_error(&self, error: String, addr: Option<SocketAddr>) {
        if let Some(addr) = addr {
            error!("网络错误 [{}]: {}", addr, error);
        } else {
            error!("网络错误: {}", error);
        }
    }
}

